package cn.ac.iie.di.wx.mq.redis;



/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import org.apache.log4j.Logger;
import redis.clients.jedis.Jedis;
import cn.ac.iie.di.wx.mq.common.RuntimeEnv;
import cn.ac.iie.di.wx.mq.utils.JedisConnectionHandler;
import cn.ac.iie.di.wx.mq.vo.RedisVo;

/**
 *
 * @author Administrator
 */
public class JedisThread implements Runnable {

    public LinkedBlockingQueue<RedisVo> redisQueue;
    private static final Logger logger = Logger.getLogger(JedisThread.class.getName());

    public JedisThread(LinkedBlockingQueue<RedisVo> redisQueue) {
        this.redisQueue = redisQueue;
    }

    @Override
    public void run() {
        Jedis jedis = null;
        JedisConnectionHandler jedisConn = null;
        try {

//            String[] ipAndPort = ((String) RuntimeEnv.getParam(RuntimeEnv.redis_hosts)).split(":");
            jedisConn = new JedisConnectionHandler((String) RuntimeEnv.getParam(RuntimeEnv.redis_hosts),
                    (String) RuntimeEnv.getParam(RuntimeEnv.redis_masterName),
                    Integer.parseInt((String) RuntimeEnv.getParam(RuntimeEnv.redis_dbName)));

//            logger.info(jedis);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        while (true) {
            try {
                jedis = jedisConn.getJedis();
                RedisVo vo = redisQueue.poll(1, TimeUnit.SECONDS);
                if (vo != null) {
                        String flag = jedis.setex(vo.getKey(), 60 * 60 * 24 * 7, vo.getValue());
                        if (flag.equals("OK")) {
                            String value = jedis.get(vo.getKey());
                            logger.info("set into redis:" + vo.getKey() + ",value = " + value);
                            if (vo.getValue().equals(value)) {
                                logger.info("set and get key = " + vo.getKey() + ";value = " + vo.getValue());
                            } else {
                                logger.warn("set and get failed,key = " + vo.getKey() + ";value = " + value);
                            }
                        } else {
                            logger.info("set " + vo.getKey() + " into redis failed flag is " + flag + ",now put into retry queue.");
                        }
                        //设置超时时间
//                        Long expire = jedis.expire(entrySet.getKey(), 60 * 60 * 24 * 7);
//                        logger.info("set expire time " + expire + ", key " + entrySet.getKey());
                        Thread.sleep(10);
                    
                }

            } catch (InterruptedException ex) {
                logger.error(ex.getMessage(), ex);
            } finally {
                if (jedis != null) {
                    jedisConn.returnResource(jedis);
                }
            }
        }
    }

}
